package com.example.starter.pubsub.redis;

import cn.hutool.core.util.ObjUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.example.starter.pubsub.annotation.RedisListener;
import com.example.starter.pubsub.api.SubListener;
import com.example.starter.util.ExceptionUtil;
import com.example.starter.util.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import sun.reflect.generics.reflectiveObjects.ParameterizedTypeImpl;

import java.lang.reflect.Type;

/**
 * @author 王令
 * @since 2023/6/30 15:20
 */
@Slf4j
public abstract class RedisSubListener<M> implements SubListener<M>, MessageListener, InitializingBean {

    private final Class<M> clazz;

    @SuppressWarnings("unchecked")
    public RedisSubListener() {
        final Type[] arguments =
                ((ParameterizedTypeImpl) this.getClass().getGenericSuperclass()).getActualTypeArguments();
        if (ObjUtil.isEmpty(arguments)) {
            throw ExceptionUtil.wrap("Please set generic");
        }
        this.clazz = (Class<M>) arguments[0];
    }

    @Override
    public final void onMessage(Message message, byte[] pattern) {
        onMessage(JSONUtil.parse(message.toString(), clazz));
    }

    @Override
    public void afterPropertiesSet() throws NoSuchMethodException {
        final RedisListener redisListener = this.getClass().getMethod("onMessage", clazz).getAnnotation(RedisListener.class);
        if (ObjUtil.isNull(redisListener) || StrUtil.isBlank(redisListener.topic())) {
            throw ExceptionUtil.wrap("Please adds the @RedisListener annotation and sets the topic on the public void onMessage(Message message) {} method.");
        }

        final ApplicationContext context = SpringUtil.getApplicationContext();

        final String topic = context.getEnvironment().resolvePlaceholders(redisListener.topic());

        final RedisMessageListenerContainer listenerContainer = context.getBean(RedisMessageListenerContainer.class);
        listenerContainer.addMessageListener(this, new ChannelTopic(topic));
        log.info("subscribe for {}", topic);
    }

}
